// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_ROW_MUTATOR_ON_TABLET_HPP_
#define _XBASE_ROW_MUTATOR_ON_TABLET_HPP_

#include <boost/function.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>
#include <boost/asio/deadline_timer.hpp>

#include "session.hpp"
#include "protocol.hpp"
#include "client_options.hpp"
#include "row_mutation_data.hpp"
#include "async_module_base.hpp"

namespace xbase
{
	class row_mutator_on_tablet
		: public async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level> base_type;
		typedef row_mutator_on_tablet this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef session session_type;
		typedef session_type::ptr_type session_ptr;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(const error_code&)> handler_type;

		typedef protocol::packet_type<protocol::WRITE_REQ>::type write_req;
		typedef protocol::packet_type<protocol::WRITE_REQ>::ptr_type write_req_ptr;

		typedef protocol::packet_type<protocol::WRITE_RESP>::type write_resp;
		typedef protocol::packet_type<protocol::WRITE_RESP>::ptr_type write_resp_ptr;

	public:
		row_mutator_on_tablet(io_service_pool &pool)
			: base_type(pool, "row_mutation_on_tablet")
			, m_options(pool.options())
			, m_retry_times(0) {}
			// , m_is_processing(false) {}

			// 向服务器会话(session)上的子表(tablet_name)提交修改
			// (mutation)，时间戳将会作为未指定时间戳的记录的时间戳。
			inline
				void async_apply(session_ptr session,
				const std::string &tablet_name,
				const row_mutation_data_ptr &mutation,
				uint64_t timestamp,
				handler_type handler)
			{
				write_req_ptr req_ptr(new write_req());
				protocol::write_request_v2 &body = req_ptr->body();
				body.m_tablet_name = tablet_name;
				body.m_mutation_data = mutation;
				body.m_timestamp = timestamp;
				//mutation.fill_record_vector(body.m_records, timestamp);

				async_apply(session, req_ptr, handler);
			}

			inline
				void async_apply(session_ptr session,
				const write_req_ptr &req_ptr, handler_type handler)
			{
				// 不能写空行！
				if(req_ptr->body().m_mutation_data->row_name().empty())
				{
					this->get_upper_level_service().dispatch(
						boost::bind(handler, error::bad_row_name)
						);
					return;
				}

				m_session = session;
				m_req_ptr = req_ptr;
				m_handler = handler;

				m_retry_times = 0;

				on_async_apply();
			}

			// template<typename Handler>
			// inline
			// bool is_processing(Handler handler) {
			// 	const bool busying = is_processing();
			// 	if(busying)
			// 	{
			// 		this->get_upper_level_service()->dispatch(
			// 			boost::bind(handler, error::busying_module, 0)
			// 			);
			// 	}
			// 	return busying;
			// }

			// bool is_processing() {
			// 	return m_is_processing;
			// }
	protected:
		// 在 lower_level_service 上执行，除第一次异步提交外
		// （为优化，直接由上层调用，而不再转到自己的
		// lower_level_service 上调用）
		//
		// 否则，在 async_apply 中应该是这样调用 on_async_apply
		// this->get_lower_level_service().dispatch(boost::bind(&this_type::on_async_apply,
		// 						        this_ptr()));
		void on_async_apply();
		// 以下函数总在 lower_level_service 上执行

		void on_response(protocol::packet_ptr ptr, const error_code &socket_ec);

		void on_retry(const error_code &ec);

		void on_error(const error_code &ec);

		void dispatch(const error_code &ec);

		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		const client_options &m_options;
		boost::shared_ptr<boost::asio::deadline_timer> m_timer;

		session_ptr m_session;
		write_req_ptr m_req_ptr;
		handler_type m_handler;

		std::size_t m_retry_times;
		// bool m_is_processing;
	};

	typedef row_mutator_on_tablet::ptr_type row_mutator_on_tablet_ptr;

	
	template<typename Handler>
	inline
	void async_apply_on_tablet(io_service_pool &pool,
				   session_ptr session,
				   const std::string &tablet_name,
				   const row_mutation_data_ptr &mutation,
				   uint64_t timestamp,
				   Handler handler) {
		row_mutator_on_tablet_ptr mutator(new_async_module(row_mutator_on_tablet, pool));
		mutator->async_apply(session, tablet_name, mutation, timestamp, handler);
	}

} // namespace xbase

#endif	// _XBASE_ROW_MUTATOR_ON_TABLET_HPP_
